package com.oildatacollection.server;

import com.oildatacollection.datareceive.RecvQueue;
import com.oildatacollection.datareceive.RecvQueueImpl;
import com.oildatacollection.datasend.SendQueue;
import com.oildatacollection.datasend.SendQueueImpl;
import com.oildatacollection.message.SendMessage;
import com.oildatacollection.message.protocol.MessageProtocolModbusImpl;
import com.oildatacollection.message.protocol.ProtocolFrameDecoder;
import com.oildatacollection.server.handler.ReceiveMessageHandler;
import com.oildatacollection.util.YamlLoader;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import java.util.*;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicBoolean;
import lombok.extern.slf4j.Slf4j;

class ChannelObj {
    private final List<Channel> chList;

    public ChannelObj() {
        chList = new ArrayList<>();
    }

    public void addChannel(Channel ch) {
        chList.add(ch);
    }

    public void closeCh() {
        for (Channel ch : chList) {
            ch.close();
        }
    }
}

class IpChanMap {
    private final Map<String, ChannelObj> ipChMap;

    public IpChanMap() {
        ipChMap = new Hashtable<>();
    }

    public boolean IpExist(String ip) {
        return ipChMap.containsKey(ip);
    }

    public void addIpCh(String ip, Channel ch) {
        if (ipChMap.containsKey(ip)) {
            if (ch != null) {
                ipChMap.get(ip).addChannel(ch);
            }
        } else {
            ChannelObj chObj = new ChannelObj();
            if (ch != null) {
                chObj.addChannel(ch);
            }
            ipChMap.put(ip, chObj);
        }
    }

    public void removeIp(String ip) {
        ChannelObj res = ipChMap.get(ip);
        if (res != null) {
            res.closeCh();
        }
        ipChMap.remove(ip);
    }
}

@Slf4j
public class NettyServerImpl implements Server {
    private int bossThreadNum;
    private int workerThreadNum;
    private String serverIp;
    private int serverPort;

    private final IpChanMap ipAllowedMap;

    private final MessageCodecSharable messageCodec;
    private final RecvQueue recvQueue;
    private final SendQueue sendQueue;
    private final ReceiveMessageHandler recvHandler;
    private Channel serverChannel;
    private final CountDownLatch cdLach;
    private final AtomicBoolean bindOver; // 是否bind完毕
    private Thread serverThread;

    public NettyServerImpl() {
        bossThreadNum = 1;
        workerThreadNum = 4;
        serverPort = 0;
        ipAllowedMap = new IpChanMap();
        messageCodec = new MessageCodecSharable();
        recvQueue = new RecvQueueImpl();
        sendQueue = new SendQueueImpl();
        recvHandler = new ReceiveMessageHandler(recvQueue, sendQueue);
        serverChannel = null;
        cdLach = new CountDownLatch(1);
        bindOver = new AtomicBoolean(false);
        serverThread = null;
    }

    /**
     * 加载配置文件 初始化参数
     */
    @SuppressWarnings("unchecked")
    private void initConfig() {
        try {
            YamlLoader yamlLoader = new YamlLoader("app-config/application-server.yml");
            Map<String, Object> serverObj = (Map<String, Object>) yamlLoader.get("server");
            bossThreadNum = (int) serverObj.get("bossThreadNum");
            workerThreadNum = (int) serverObj.get("workerThreadNum");
            serverIp = (String) serverObj.get("ip");
            serverPort = (int) serverObj.get("port");
        } catch (RuntimeException e) {
            log.error("初始化server失败", e);
        }
        log.trace("server配置初始化成功:" +
                " bossThreadNum=" + bossThreadNum +
                " workerThreadNum=" + workerThreadNum +
                " ip=" + serverIp +
                " port=" + serverPort);
    }

    private void runInThread() {
        initConfig();
        // 开启发送与接收队列
        recvQueue.start();
        sendQueue.start();

        NioEventLoopGroup boss = new NioEventLoopGroup(bossThreadNum);
        NioEventLoopGroup worker = new NioEventLoopGroup(workerThreadNum);
        LoggingHandler LOGGING_HANDLER = new LoggingHandler(LogLevel.DEBUG);
        try {
            // 初始化服务器
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap.channel(NioServerSocketChannel.class);
            serverBootstrap.group(boss, worker);
            serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel ch) throws Exception {
                    // 保存当前连接的套接字
                    String ip = ch.remoteAddress().getHostName();
                    if (!ipAllowedMap.IpExist(ip)) {
                        log.trace("非法IP(" + ip + ")访问");
                        return;
                    } else {
                        ipAllowedMap.addIpCh(ip, ch);
                    }
                    int port = ch.remoteAddress().getPort();
                    log.trace("新的socket连接: " + ip + ":" + port);

                    int masMessSize = MessageProtocolModbusImpl.getMaxMessageSize();
                    int offSet = MessageProtocolModbusImpl.getLengthFieldOffset();
                    int lenFieldLen = MessageProtocolModbusImpl.getLengthFieldLength();
                    ch.pipeline().addLast(new ProtocolFrameDecoder(masMessSize, offSet, lenFieldLen));
                    ch.pipeline().addLast(LOGGING_HANDLER);
                    ch.pipeline().addLast(messageCodec);
                    ch.pipeline().addLast(recvHandler);
                }
            });
            serverChannel = serverBootstrap.bind(serverIp, serverPort).sync().channel();
            cdLach.countDown();
            bindOver.set(true);
            serverChannel.closeFuture().sync();
        } catch (Exception e) {
            log.error("服务器错误", e);
        } finally {
            boss.shutdownGracefully();
            worker.shutdownGracefully();
            recvQueue.stop();
            sendQueue.stop();
        }
    }

    @Override
    public void start() {
        ipAllowedMap.addIpCh("127.0.0.1", null);
        try {
            if (serverThread != null) {
                serverThread.join();
            }
            serverThread = new Thread(new Runnable() {
                @Override
                public void run() {
                    runInThread();
                }
            });
            serverThread.start();
            cdLach.await();
        } catch (InterruptedException e) {
            log.error("server线程错误", e);
        }
    }

    @Override
    public void sendMessage(int wellID, byte[] msg) {
        sendQueue.sendToQueue(new SendMessage(wellID, msg));
    }

    @Override
    public void changeIpSet(String ip, boolean state) {
        if (state) {
            ipAllowedMap.addIpCh(ip, null);
            log.trace("ip:" + ip + "已添加至白名单");
        } else {
            ipAllowedMap.removeIp(ip);
            log.trace("ip" + ip + "已从白名单中删除");
        }
    }

    @Override
    public void close() {
        if (!bindOver.get()) {
            log.error("正在初始化中，尚不能关闭服务器");
            return;
        }
        serverChannel.close();
    }

    @Override
    public void waitForEnd() {
        if (serverThread != null) {
            try {
                serverThread.join();
            } catch (InterruptedException e) {
                log.error("server关闭时错误", e);
            }
        }
    }
}
